package com.shujia.flink.tf;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

public class Demo06Window {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(4);

        DataStreamSource<String> socketDS = env.socketTextStream("master", 8888);

        socketDS.flatMap((line, out) -> {
                    for (String word : line.split(",")) {
                        out.collect(word);
                    }
                }, Types.STRING)
                .map(word -> Tuple2.of(word, 1), Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(t2 -> t2.f0, Types.STRING)
                // 基于时间的窗口还会分为：处理时间ProcessTime、事件时间EventTime
                // 基于时间的窗口：滑动、滚动
                // 每隔5s 统计 最近 10s
//                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                // 每隔5s 统计 最近 5s
//                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                // 基于数量的窗口：计数窗口
                // 滑动、滚动
                // 滚动的计数窗口：每5条数据统计一次
//                .countWindow(5)
                // 滑动的计数窗口：每5条统计最近10条数据的情况
//                .countWindow(10, 5)
                // Session会话窗口：当多久，例如5s没有数据 则会触发窗口的统计
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
                .sum(1)
                .print();

        env.execute();


    }
}
